package events

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/IBM/sarama"
	"go.uber.org/zap"
	"jk-time/webook/pkg/logger"
	"jk-time/webook/pkg/saramax"
	"jk-time/webook/search/service"
	"time"
)

type InteractiveEvent struct {
	Type  int32  `json:"type,omitempty"`
	Uid   int32  `json:"uid"`
	Biz   string `json:"biz"`
	BizId int32  `json:"biz_id"`
}

func (InteractiveEvent) Topic() string {
	return "search_interactives"
}

type InteractiveKafkaComsumer struct {
	client   sarama.Client
	l        logger.Logger
	syncSvc  service.SyncService
	handlers map[int32]InteractiveHandler
}

func NewInteractiveConsumer(client sarama.Client, l logger.Logger, svc service.SyncService) *InteractiveKafkaComsumer {
	handlers := map[int32]InteractiveHandler{
		1: &LikeHandler{syncSvc: svc},
		2: &CollectHandler{syncSvc: svc},
		3: &CancelLikeHandler{syncSvc: svc},
	}
	return &InteractiveKafkaComsumer{
		syncSvc:  svc,
		client:   client,
		l:        l,
		handlers: handlers,
	}
}
func NewInteractiveKafkaComsumer(client sarama.Client, l logger.Logger, syncSvc service.SyncService) Consumer {
	return &InteractiveKafkaComsumer{
		client:  client,
		syncSvc: syncSvc,
		l:       l,
	}
}
func (k *InteractiveKafkaComsumer) Start() error {
	cg, err := sarama.NewConsumerGroupFromClient("search_interactive_1", k.client)
	if err != nil {
		return err
	}
	go func() {
		err := cg.Consume(context.Background(),
			[]string{InteractiveEvent{}.Topic()},
			saramax.NewHandler[InteractiveEvent]("search_interactive_1", k.Consume))
		if err != nil {
			zap.L().Error("退出了消费循环,异常", zap.Error(err))
		}
	}()
	return err
}

func (k *InteractiveKafkaComsumer) Consume(msg *sarama.ConsumerMessage, t InteractiveEvent) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	handleFunc, ok := k.handlers[t.Type]
	if !ok {
		k.l.Error("未知类型")
		return nil
	}
	return handleFunc.Handle(ctx, t)
}
func getDocId(data InteractiveEvent) string {
	return fmt.Sprintf("%d_%s_%d", data.Uid, data.Biz, data.BizId)
}

type InteractiveHandler interface {
	Handle(ctx context.Context, data InteractiveEvent) error
}
type LikeHandler struct {
	syncSvc service.SyncService
}

func (l *LikeHandler) Handle(ctx context.Context, data InteractiveEvent) error {
	return handle(ctx, l.syncSvc, "like_idx", getDocId(data), data)
}

type CollectHandler struct {
	syncSvc service.SyncService
}

func (c *CollectHandler) Handle(ctx context.Context, data InteractiveEvent) error {
	return handle(ctx, c.syncSvc, "collect_idx", getDocId(data), data)
}

type CancelLikeHandler struct {
	syncSvc service.SyncService
}

func (c *CancelLikeHandler) Handle(ctx context.Context, data InteractiveEvent) error {
	return c.syncSvc.Delete(ctx, "like_idx", getDocId(data))
}
func handle(ctx context.Context, syncSvc service.SyncService, index, key string, data any) error {
	val, err := json.Marshal(data)
	if err != nil {
		return err
	}
	return syncSvc.InputAny(ctx, index, key, string(val))
}
